package org.cainiao.basic.reactive.error;

import org.cainiao.basic.reactive.error.exception.BusinessException;
import org.cainiao.basic.reactive.error.exception.UnknownKeyException;
import org.cainiao.basic.reactive.error.util.MyWrapper;
import org.cainiao.basic.reactive.error.util.Value;
import org.cainiao.basic.reactive.mock.Assert;
import org.cainiao.basic.reactive.mock.MyHttpRequest;
import org.cainiao.basic.reactive.mock.SomeAutoCloseable;
import org.cainiao.basic.reactive.mock.Stats;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

public class ErrorTest {

    public static void main(String[] args) throws InterruptedException {
        ErrorTest errorTest = new ErrorTest();
        errorTest.test();
    }

    MyHttpRequest httpRequest = new MyHttpRequest();

    public void test() {
        AtomicInteger errorCount = new AtomicInteger();
        Flux<Integer> transientFlux = httpRequest.get()
            .doOnError(e -> errorCount.incrementAndGet());

        transientFlux.retryWhen(Retry.max(2).transientErrors(true))
            .blockLast();
        assertThat(errorCount).hasValue(6);
    }

    private Assert assertThat(AtomicInteger errorCount) {
        System.out.println(errorCount);
        return new Assert(errorCount);
    }

    public void test25() throws InterruptedException {
        Flux<String> flux = Flux.interval(Duration.ofMillis(250))
            .map(input -> {
                if (input < 3) {
                    return "tick " + input;
                }
                throw new RuntimeException("boom");
            })
            .retryWhen(Retry.from(companion ->
                companion.map(rs -> {
                    System.out.printf("rs.totalRetries() = %d\n", rs.totalRetries());
                    if (rs.totalRetries() < 3) {
                        return rs.totalRetries();
                    }
                    throw new RuntimeException("boom 2");
                })
            ));
        flux.elapsed().subscribe(str -> System.out.println("sub ->> " + str), System.err::println);
        Thread.sleep(4100);
    }

    public void test24() throws InterruptedException {
        Flux<String> flux = Flux.interval(Duration.ofMillis(250))
            .map(input -> {
                if (input < 3) {
                    return "tick " + input;
                }
                throw new RuntimeException("boom");
            })
            .retryWhen(Retry.from(companion ->
                companion.take(1).map(rs -> {
                    System.out.printf("rs.totalRetries() = %d\n", rs.totalRetries());
                    return rs.totalRetries();
                })));
        flux.elapsed().subscribe(str -> System.out.println("sub ->> " + str), System.err::println);
        Thread.sleep(4100);
    }

    public void test23() throws InterruptedException {
        Flux<String> flux = Flux.interval(Duration.ofMillis(250))
            .map(input -> {
                if (input < 3) {
                    return "tick " + input;
                }
                throw new RuntimeException("boom");
            })
            .retry(3);
        flux.elapsed().subscribe(str -> System.out.println("sub ->> " + str), System.err::println);
        Thread.sleep(4100);
    }

    public void test22() throws InterruptedException {
        Flux<String> flux = Flux.interval(Duration.ofMillis(250))
            .map(input -> {
                if (input < 3) {
                    return "tick " + input;
                }
                throw new RuntimeException("boom");
            })
            .onErrorReturn("Uh oh");
        flux.subscribe(System.out::println);
        Thread.sleep(2100);
    }

    public void test1() {
        Flux.just(1, 2, 0)
            .map(i -> "100 / " + i + " = " + (100 / i)) // this triggers an error with 0
            .onErrorReturn("Divided by zero :("); // error handling example
    }

    public void test2() {
        Flux<String> s = Flux.range(1, 10)
            .map(this::doSomethingDangerous)
            .map(this::doSecondTransform);
        s.subscribe(value -> System.out.println("RECEIVED " + value),
            error -> System.err.println("CAUGHT " + error)
        );
    }

    public void test3() {
        try {
            for (int i = 1; i < 11; i++) {
                String v1 = doSomethingDangerous(i);
                String v2 = doSecondTransform(v1);
                System.out.println("RECEIVED " + v2);
            }
        } catch (Throwable t) {
            System.err.println("CAUGHT " + t);
        }
    }

    public String test4() {
        try {
            return doSomethingDangerous(10);
        } catch (Throwable error) {
            return "RECOVERED";
        }
    }

    public void test5() {
        Flux.just(10)
            .map(this::doSomethingDangerous)
            .onErrorReturn("RECOVERED");
    }

    public void test6() {
        Flux.just(10)
            .map(this::doSomethingDangerous)
            .onErrorReturn(e -> e.getMessage().equals("boom10"), "recovered10");
    }

    public void test7() {
        Flux.just(10, 20, 30)
            .map(this::doSomethingDangerousOn30)
            .onErrorComplete();
    }

    public void test8() {
        String v1;
        try {
            v1 = callExternalService2("key1");
        } catch (Throwable error) {
            v1 = getFromCache2("key1");
        }
        String v2;
        try {
            v2 = callExternalService2("key2");
        } catch (Throwable error) {
            v2 = getFromCache2("key2");
        }
    }

    public void test9() {
        Flux.just("key1", "key2")
            .flatMap(k -> callExternalService(k)
                .onErrorResume(e -> getFromCache(k))
            );
    }

    public void test10() {
        Flux.just("timeout1", "unknown", "key2")
            .flatMap(k -> callExternalService(k)
                .onErrorResume(error -> {
                    if (error instanceof TimeoutException) {
                        return getFromCache(k);
                    }
                    if (error instanceof UnknownKeyException) {
                        return registerNewEntry(k, "DEFAULT");
                    }
                    return Flux.error(error);
                })
            );
    }

    public Object test11() {
        try {
            Value v = erroringMethod();
            return MyWrapper.fromValue(v);
        } catch (Throwable error) {
            return MyWrapper.fromError(error);
        }
    }

    private final Flux<String> erroringFlux = Flux.just("test");

    public void test12() {
        erroringFlux.onErrorResume(error -> Mono.just(MyWrapper.fromError(error)));
    }

    public Flux<Value> test13(Value k) {
        try {
            return callExternalService(k);
        } catch (Throwable error) {
            throw new BusinessException("oops, SLA exceeded", error);
        }
    }

    public void test14() {
        Flux.just("timeout1")
            .flatMap(this::callExternalService)
            .onErrorResume(original -> Flux.error(new BusinessException("oops, SLA exceeded", original)));
    }

    public void test15() {
        Flux.just("timeout1")
            .flatMap(this::callExternalService)
            .onErrorMap(original -> new BusinessException("oops, SLA exceeded", original));
    }

    public Flux<Value> test16(Value k) {
        try {
            return callExternalService(k);
        } catch (RuntimeException error) {
            // make a record of the error
            log("uh oh, falling back, service failed for key " + k);
            throw error;
        }
    }

    public void test17() {
        LongAdder failureStat = new LongAdder();
        Flux<String> flux = Flux.just("unknown")
            .flatMap(k -> callExternalService(k)
                .doOnError(e -> {
                    failureStat.increment();
                    log("uh oh, falling back, service failed for key " + k);
                })
            );
    }

    public void test18() {
        Stats stats = new Stats();
        stats.startTimer();
        try {
            doSomethingDangerous();
        } finally {
            stats.stopTimerAndRecordTiming();
        }
    }

    public String test19() {
        try (SomeAutoCloseable disposableInstance = new SomeAutoCloseable()) {
            return disposableInstance.toString();
        }
    }

    public void test20() {
        Stats stats = new Stats();
        LongAdder statsCancel = new LongAdder();
        Flux<String> flux = Flux.just("foo", "bar")
            .doOnSubscribe(s -> stats.startTimer())
            .doFinally(type -> {
                stats.stopTimerAndRecordTiming();
                if (type == SignalType.CANCEL) {
                    statsCancel.increment();
                }
            })
            .take(1);
    }

    private final AtomicBoolean isDisposed = new AtomicBoolean();
    private final Disposable disposableInstance = new Disposable() {
        @Override
        public void dispose() {
            isDisposed.set(true);
        }

        @Override
        public String toString() {
            return "DISPOSABLE";
        }
    };

    public void test21() {
        Flux<String> flux = Flux.using(
            () -> disposableInstance,
            disposable -> Flux.just(disposable.toString()),
            Disposable::dispose
        );
    }

    private void log(String message) {
        System.out.println(message);
    }

    private Value erroringMethod() {
        return new Value();
    }

    private <T> Flux<T> callExternalService(T v) {
        return Flux.just(v);
    }

    private String callExternalService2(String v) {
        return v;
    }

    private String getFromCache2(String v) {
        return String.valueOf(v);
    }

    private Publisher<String> getFromCache(String v) {
        return Flux.just(v);
    }

    private Publisher<String> registerNewEntry(String v, String type) {
        System.out.println(type);
        return Flux.just(v);
    }

    private void doSomethingDangerous() {
        System.out.println("doSomethingDangerous");
    }

    private String doSomethingDangerous(Integer v) {
        return String.valueOf(v);
    }

    private String doSecondTransform(String v) {
        return v;
    }

    private String doSomethingDangerousOn30(Integer v) {
        return String.valueOf(v);
    }


}
